package com.dataaccess.dataaccess.config.rocketmq;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.dataaccess.dataaccess.support.udp.DataReceiveConfigration;
import com.dataaccess.dataaccess.support.utils.MyLogger;
import com.dataaccess.dataaccess.web.service.*;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * @Author 18011618
 * @Date 10:31 2018/7/18
 * @Function 模拟用户消息消费
 */
@Component
public class UserConsumerBackups {
    /**
     * 消费者的组名
     */
    @Value("${suning.rocketmq.conumerGroupBackups}")
    private String consumerGroup;

    /**
     * NameServer 地址
     */
    @Value("${suning.rocketmq.namesrvaddr}")
    private String namesrvAddr;

    @Resource
    public MyLogger logger;

    @Resource
    public DeviceAboatCombustiblegasService deviceAboatCombustiblegasService;
    @Resource
    public DeviceAboatElectricService deviceAboatElectricService;
    @Resource
    private DeviceAboatFireExtinguisherService deviceAboatFireExtinguisherService;
    @Resource
    public DeviceAboatFirewaterService deviceAboatFirewaterService;
    @Resource
    public DeviceAboatSmokeService deviceAboatService;
    @Resource
    private VisualizationVideoAlarmService videoAlarmService;
    @Resource
    DataReceiveConfigration dataReceiveConfigration;
    @Resource
    PostOrGetService postOrGetService;

    //@PostConstruct
    public void consumer() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);
        try {
            consumer.subscribe("user-topic", "user-tag");
            //注册消息监听器处理消息
//            consumer.registerMessageListener(new MessageListenerConcurrently() {
//                @Override
//                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context){
//                    //对消息进行处理
//                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//                }
//            });
            consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {
                try {
                    for (MessageExt messageExt : list) {
                        String mes = new String(messageExt.getBody());
                        logger.info("consumerBackups"+"------"+mes);
                        JSONObject jsonObject = JSONUtil.parseObj(mes);
                        String deviceFlag = jsonObject.getStr("deviceFlag");
                        if(deviceFlag.equals("udpCombustiblegas")){
                            dataReceiveConfigration.handleMes(mes,"udp");
                        }else if(deviceFlag.equals("combustiblegas")){
                            deviceAboatCombustiblegasService.deviceDataChanged(jsonObject);
                        }else if(deviceFlag.equals("electric")){
                            try{
                                deviceAboatElectricService.deviceDataChanged(jsonObject);
                            }catch (Exception e){
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("fireExtinguisher")){
                            deviceAboatFireExtinguisherService.deviceDataChanged(jsonObject);
                        }else if(deviceFlag.equals("firewater")){
                            try{
                                deviceAboatFirewaterService.deviceDataChanged(jsonObject);
                            }catch (Exception e){
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("smoke")){
                            deviceAboatService.deviceDataChanged(jsonObject);
                        }else if(deviceFlag.equals("video")){
                            try{
                                videoAlarmService.fluoriteCloud(jsonObject);
                            }catch (Exception e){
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("aliyunCamera")){
                            try{
                                videoAlarmService.aliyunCamera(jsonObject);
                            }catch (Exception e){
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("updateToOff")){
                            try{
                                postOrGetService.updateToOff(jsonObject);
                            }catch (Exception e){
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("test")){
                            System.err.println("消费消息: " + "deviceFlag:"+deviceFlag+"-----"+mes);
                        }

                        //System.err.println("消费消息: " + new String(messageExt.getBody()));//输出消息内容
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍后再试
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功
            });
            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
